import os
import subprocess
import sys
import time
from data_generator import KafkaProducer


def execute(prompt: str, command: str):
    print(prompt)
    print(command)
    return subprocess.Popen(command, shell=True)


class KafkaLauncher:
    def __init__(self, kafka_path: str = None,
                 python_interpreter: str = None):
        self.kafka_path = kafka_path
        if kafka_path is None:
            with open(os.path.join("conf", "kafka.path.txt")) as f:
                self.kafka_path = f.read()

        self.python_interpreter = python_interpreter \
            if python_interpreter is not None \
            else ("python3" if sys.platform == "Windows" else "python" )

        self.script_dir = os.path.join(self.kafka_path, "bin")
        self.configuration_dir=os.path.join(self.kafka_path,"config")
        self.script_dir = os.path.join(self.script_dir, "windows")
        print("init kafka launcher")
        self.producer=KafkaProducer()
    
    def produce_data(self):
        self.producer.send_test_data()

    def start_kafka(self):
        os.chdir(self.kafka_path)
        self.start_zookeeper()
        kafka_config=os.path.join(self.configuration_dir,"server.properties")
        kafka_start_script=os.path.join(self.script_dir,"kafka-server-start.bat")
        kafka_cmd=f"{kafka_start_script} {kafka_config}"
        execute("Kafka...",kafka_cmd)
        os.system("pause")
        time.sleep(10)

    def start_zookeeper(self):
        zookeeper_launcher = os.path.join(self.script_dir, "zookeeper-server-start.bat")
        zookeeper_config=os.path.join(self.configuration_dir,"zookeeper.properties")
        zookeeper_cmd = f"{zookeeper_launcher} {zookeeper_config}"
        execute("Zookeeper...",zookeeper_cmd)
        time.sleep(10)

if __name__ == '__main__':
    launcher=KafkaLauncher()
    launcher.start_kafka()
